package com.smile.frame.mq.rocket.producer;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQUtil;
import org.springframework.messaging.Message;
import org.springframework.messaging.converter.StringMessageConverter;

/**
 * 发送事务消息 - 商品添加到购物车
 * rocketMQTemplateBeanName 指定 rocketMQTemplate
 * 有多个事务事件，需要重新定义一个新的rocketMQTemplate ：如 ExtRocketMQTemplate
 *
 * @author LGC
 */
@Slf4j
@RocketMQTransactionListener(rocketMQTemplateBeanName = "rocketMQTemplate")
public class CartItemAddTransactionListener implements RocketMQLocalTransactionListener {

    /**
     * 本地事务执行会有三种可能
     * 1、commit 成功
     * 2、rollback 失败
     * 3、unknow 网络等原因 blocker还未知事务状态
     * <p>
     * COMMIT_MESSAGE 经过blocker二次确认，本地事务已提交，消费者可以去消费
     * ROLLBACK_MESSAGE 本地事务异常回滚，不提交本地事务，删除blocker消息，消费者不会去消费
     * UNKNOW 状态未知，本地事务回查，重新确认事务状态
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        org.apache.rocketmq.common.message.Message message = RocketMQUtil.convertToRocketMessage(new StringMessageConverter(), "UTF-8", arg.toString(), msg);
        String transactionId = (String) msg.getHeaders().get("rocketmq_TRANSACTION_ID");
        log.info("msg:{}", msg);
        log.info("message:{}", message);
        log.info("arg:{}", arg);
        log.info("transactionId:{}", transactionId);


        // 创建订单。。。 根据返回结果返回 RocketMQLocalTransactionState
//        try {
//            String body = new String(message.getBody());
//            OrderDTO order = JSONObject.parseObject(body, OrderDTO.class);
//            int result = orderService.createOrder(order, message.getTransactionId());
//            if (result == 1) {
//                state = LocalTransactionState.COMMIT_MESSAGE;
//                log.info("本地事务已提交。{}", message.getTransactionId());
//            }
//            if (result == 0) {
//                state = LocalTransactionState.ROLLBACK_MESSAGE;
//                log.info("本地事务未提交。{}", message.getTransactionId());
//            }
//
//        } catch (Exception e) {
//            log.info("执行本地事务失败。{0}", e);
//            state = LocalTransactionState.ROLLBACK_MESSAGE;
//        }

        return RocketMQLocalTransactionState.COMMIT;
    }

    /**
     * 本地事务进行回查，只有 LocalTransactionState.UNKNOW 才会进行回查，重新确认事务状态
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        // SpringBoot的消息对象中，并没有transactionId这个属性。跟原生API不一样。
        String transactionId = (String) msg.getHeaders().get("rocketmq_TRANSACTION_ID");

        log.info("开始回查本地事务状态。{}", transactionId);

        log.info("checkLocalTransaction msg:{}", msg);
        RocketMQLocalTransactionState state = RocketMQLocalTransactionState.COMMIT;

//        LambdaQueryWrapper<TransactionLog> queryWrapper = new LambdaQueryWrapper<>();
//        queryWrapper.eq(TransactionLog::getId, transactionId);
//        if (transactionLogMapper.selectCount(queryWrapper) > 0) {
//            state = RocketMQLocalTransactionState.COMMIT;
//        } else {
//            state = RocketMQLocalTransactionState.UNKNOWN;
//        }
        log.info("结束本地事务状态查询：{}", state);
        return state;
    }
}
